package com.lybus.lyxackunimall.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.Descriptors;
import com.google.protobuf.InvalidProtocolBufferException;
import com.lybus.lyxackunimall.db.LyxDBService;
import com.lybus.lyxackunimall.db.UnimallDbService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * canal 客户端
 */
@Service
public class ClientWarp implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    private CanalConnector canalConnector;

    @Autowired
    private LyxDBService lyxDBService;

    @Autowired
    private UnimallDbService unimallDbService;

    @Autowired
    private ApplicationContext applicationContext;

    @Value(value = "app.canalHost")
    private String host;

    // 已开始
    private volatile boolean started = false;

    private Logger logger = LoggerFactory.getLogger(getClass());


    private void start(){
        canalConnector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(
                        host, 11111
                ),
                "example",
                "",
                ""

        );
        canalConnector.connect();
        canalConnector.subscribe();
        started = true;

    }

    public void run(){
        ClientWarp bean = applicationContext.getBean(ClientWarp.class);
        try {
            while (started){
                if(bean.doRun() == 0){
                    Thread.sleep(2000);
                }

            }

        }
       catch (Exception e){
           logger.warn("canal 转换失败", e);
       }
        // 已结束
        if(!started){
            return;
        }
        if(canalConnector !=null && !canalConnector.checkValid()){
            try{
                canalConnector.disconnect();
            }catch (Exception e){
                logger.warn("canal 关闭失败", e);
            }finally {
                canalConnector = null;
                while (started){
                    try{
                        start();
                        break;
                    }catch (Exception e){
                        logger.warn("启动canal 异常",e);
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException ex) {
                            //
                        }
                    }
                }
                run();
            }

        }else {
            run();
        }

    }
    @Transactional
    public int doRun(){
        Message message = canalConnector.getWithoutAck(50);
        long batchId = message.getId();
        List<CanalEntry.Entry> entries = message.getEntries();
        LocalDateTime last = LocalDateTime.of(2021,11,30, 0,0,0);
        for (CanalEntry.Entry entry : entries) {
            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {

            }
            String schemaName = entry.getHeader().getSchemaName();
            String tableName = entry.getHeader().getTableName();
            CanalEntry.EventType eventType = rowChange.getEventType();
            if(CanalEntry.EventType.INSERT.equals(eventType)){
                if("lyx_app".equalsIgnoreCase(schemaName) && "sys_user".equalsIgnoreCase(tableName)){

                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        String phone = null; Long lyxId = null; String password = null; String salt = null;
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        for (CanalEntry.Column column : afterColumnsList) {
                            if(column.getName().equalsIgnoreCase("user_id")){
                                lyxId = Long.parseLong(column.getValue());
                            }else if(column.getName().equalsIgnoreCase("phonenumber")){
                                phone = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("password")){
                                password = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("salt")){
                                salt = column.getValue();
                            }
                        }
                        if(StringUtils.hasText(phone)) {
                            unimallDbService.insertUser(phone, lyxId, password, salt);
                            logger.warn("数据同步: phone:{}, lyxId:{}", phone, lyxId);
                        }else {
                            logger.warn("数据同步: 未找到手机号, lyxId:{}", lyxId);
                        }

                    }

                }
                else if("unimall".equalsIgnoreCase(schemaName) && "unimall_user".equalsIgnoreCase(tableName)){

                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        String phone = null; Long unimallId = null; String password = null; String salt = null;
                        String gmt_create = null;
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        for (CanalEntry.Column column : afterColumnsList) {
                            if(column.getName().equalsIgnoreCase("id")){
                                unimallId = Long.parseLong(column.getValue());
                            }else if(column.getName().equalsIgnoreCase("phone")){
                                phone = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("password")){
                                password = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("salt")){
                                salt = column.getValue();
                            }
                            else if(column.getName().equalsIgnoreCase("gmt_create")){
                                gmt_create = column.getValue();
                            }
                        }

                        if(StringUtils.hasText(phone)
                                &&
                                (
                                        (StringUtils.hasText(gmt_create) && LocalDateTime.parse(gmt_create, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")).isAfter(last))
                                        || !StringUtils.hasText(gmt_create)
                                )

                        ){

                            lyxDBService.insertUser(phone, unimallId, password, salt);
                            logger.info("数据同步: phone:{}, unimallId:{}", phone, unimallId);

                        }else {
                            logger.warn("数据同步err: phone:{}, unimallId:{}", phone, unimallId);
                        }

                    }

                }
                else {
                    logger.warn("数据新增未知: schemaName:{}, tableName:{}", schemaName, tableName);
                }
            }else if(CanalEntry.EventType.UPDATE.equals(eventType)){
                if("lyx_app".equalsIgnoreCase(schemaName) && "sys_user".equalsIgnoreCase(tableName)){
                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        String phone = null; Long lyxId = null; String password = null; String salt = null;
                        String oldPhone = null; Long oldLyxId = null; String oldPassword = null; String oldSalt = null;
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        for (CanalEntry.Column column : afterColumnsList) {
                            if(column.getName().equalsIgnoreCase("user_id")){
                                lyxId = Long.parseLong(column.getValue());
                            }else if(column.getName().equalsIgnoreCase("phonenumber")){
                                phone = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("password")){
                                password = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("salt")){
                                salt = column.getValue();
                            }
                        }
                        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                        for (CanalEntry.Column column : beforeColumnsList) {
                            if(column.getName().equalsIgnoreCase("user_id")){
                                oldLyxId = Long.parseLong(column.getValue());
                            }else if(column.getName().equalsIgnoreCase("phonenumber")){
                                oldPhone = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("password")){
                                oldPassword = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("salt")){
                                oldSalt = column.getValue();
                            }
                        }
                        if(!Objects.equals(oldPhone, phone) ||
                                !Objects.equals(oldPassword, password) ||
                                !Objects.equals(oldSalt, salt)
                        ){
                            if(StringUtils.hasText(phone)) {
                                logger.info("数据更新: phone:{}, lyxId:{}", phone, lyxId);
                                unimallDbService.updateUser(phone, lyxId, password, salt, oldPhone);
                            }else {
                                logger.warn("数据更新err: phone:{}, lyxId:{}", phone, lyxId);
                            }
                        }
                    }
                }
                else if("unimall".equalsIgnoreCase(schemaName) && "unimall_user".equalsIgnoreCase(tableName)){

                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        String phone = null; Long unimallId = null; String password = null; String salt = null;
                        String gmt_create = null;
                        String oldPhone = null; Long oldUnimallId = null; String oldPassword = null; String oldSalt = null;
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        for (CanalEntry.Column column : afterColumnsList) {
                            if(column.getName().equalsIgnoreCase("id")){
                                unimallId = Long.parseLong(column.getValue());
                            }else if(column.getName().equalsIgnoreCase("phone")){
                                phone = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("password")){
                                password = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("salt")){
                                salt = column.getValue();
                            } else if(column.getName().equalsIgnoreCase("gmt_create")){
                                gmt_create = column.getValue();
                            }
                        }
                        for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                            if(column.getName().equalsIgnoreCase("id")){
                                oldUnimallId = Long.parseLong(column.getValue());
                            }else if(column.getName().equalsIgnoreCase("phone")){
                                oldPhone = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("password")){
                                oldPassword = column.getValue();
                            }else if(column.getName().equalsIgnoreCase("salt")){
                                oldSalt = column.getValue();
                            }
                        }
                        if(!Objects.equals(oldPhone, phone) ||
                                !Objects.equals(oldPassword, password) ||
                                !Objects.equals(oldSalt, salt)
                        ){
                            if(StringUtils.hasText(phone)
                                    &&
                                    (
                                            (StringUtils.hasText(gmt_create) && LocalDateTime.parse(gmt_create, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")).isAfter(last))
                                                    || !StringUtils.hasText(gmt_create)
                                    )

                            ){
                                lyxDBService.updateUser(phone, unimallId, password, salt);
                                logger.info("数据更新: phone:{}, unimallId:{}", phone, unimallId);
                            }else {
                                logger.warn("数据更新err: phone:{}, unimallId:{}", phone, unimallId);
                            }


                        }else {
                            logger.warn("数据更新一致: phone:{}, unimallId:{}", phone, unimallId);
                        }
                    }

                }
                else {
                    logger.warn("数据更新未知: schemaName:{}, tableName:{}", schemaName, tableName);
                }
            }

        }
        if(batchId > 0){
            canalConnector.ack(batchId);
        }

        logger.info("数据同步: batchId:{}, 数量:{}" , batchId, entries.size());
        return entries.size();
    }


    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if(event.getApplicationContext().getParent() == null){
            applicationContext = event.getApplicationContext();
            start();
            run();

        }
    }

    @Override
    public void destroy() throws Exception {
        try{
            if(canalConnector != null && canalConnector.checkValid()){
                canalConnector.disconnect();
                canalConnector = null;
            }
        }catch (Exception exception){

        }

        started = false;
    }
}
